//
// Copyright (C) 2024 Kumo inc.
// Author: Jeff.li lijippy@163.com
// All rights reserved.
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as published
// by the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// This program is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
// GNU Affero General Public License for more details.
//
// You should have received a copy of the GNU Affero General Public License
// along with this program.  If not, see <https://www.gnu.org/licenses/>.
//



#include <turbo/log/logging.h>
#include <turbo/strings/ascii.h>
#include <turbo/flags/flag.h>
#include <turbo/bootstrap/servlet.h>
#include <turbo/log/logging.h>
#include <hadarservice/service.h>
#include <hadarservice/proto/api.pb.h>
#include <krpc/rpc/channel.h>

TURBO_FLAG(std::string, protocol, "baidu_std", "Protocol type. Defined in krpc/rpc/options.proto");
TURBO_FLAG(std::string, connection_type, "", "Connection type. Available values: single, pooled, short");
TURBO_FLAG(std::string, server, "0.0.0.0:8801", "IP Address of server");
TURBO_FLAG(std::string, load_balancer, "", "The algorithm for load balancing");
TURBO_FLAG(int32_t, timeout_ms, 100, "RPC timeout in milliseconds");
TURBO_FLAG(int32_t, max_retry, 3, "Max retries(not including the first RPC)");
TURBO_FLAG(int32_t, interval_ms, 1000, "Milliseconds between consecutive requests");
TURBO_FLAG(std::string, convert_type, "s2t", "convert type");
TURBO_FLAG(std::string, content, "s2t", "convert type");

int main(int argc, char* argv[]) {
    auto &servlet = turbo::Servlet::instance();
    auto *app = servlet.run_app();
    app->add_option("-p,--protocol", FLAGS_protocol, "Protocol type. Defined in krpc/rpc/options.proto");
    app->add_option("-C,--connection_type", FLAGS_connection_type, "Connection type. Available values: single, pooled, short");
    app->add_option("-s,--server", FLAGS_server, "IP Address of server");
    app->add_option("-l,--load_balancer", FLAGS_load_balancer, "The algorithm for load balancing");
    app->add_option("-T,--timeout_ms", FLAGS_timeout_ms, "RPC timeout in milliseconds");
    app->add_option("-m,--max_retry", FLAGS_timeout_ms, "Max retries(not including the first RPC)");
    app->add_option("-i,--interval_ms", FLAGS_interval_ms, "Milliseconds between consecutive requests");
    app->add_option("-t,--type", FLAGS_convert_type, "Milliseconds between consecutive requests")->required();
    app->add_option("-c,--content", FLAGS_content, "Milliseconds between consecutive requests")->required();

    auto [ext, code] = servlet.run(argc, argv);
    if(ext) {
        return code;
    }

    // A Channel represents a communication line to a Server. Notice that
    // Channel is thread-safe and can be shared by all threads in your program.
    krpc::Channel channel;

    // Initialize the channel, nullptr means using default options.
    krpc::ChannelOptions options;
    options.protocol = turbo::get_flag(FLAGS_protocol);
    options.connection_type = turbo::get_flag(FLAGS_connection_type);
    options.timeout_ms = turbo::get_flag(FLAGS_timeout_ms)/*milliseconds*/;
    options.max_retry = turbo::get_flag(FLAGS_max_retry);
    if (channel.Init(turbo::get_flag(FLAGS_server).c_str(), turbo::get_flag(FLAGS_load_balancer).c_str(), &options) != 0) {
        LOG(ERROR) << "Fail to initialize channel";
        return -1;
    }

    // Normally, you should not call a Channel directly, but instead construct
    // a stub Service wrapping it. stub can be shared by all threads as well.
    hadar_service::convert_Stub stub(&channel);
    std::string lc_convert_type = turbo::str_to_lower(turbo::get_flag(FLAGS_convert_type));
    int log_id = 0;
    hadar_service::HadarRequest request;
    hadar_service::HadarResponse response;
    krpc::Controller cntl;
    request.set_message(turbo::get_flag(FLAGS_content));
    cntl.set_log_id(log_id ++);

    if(lc_convert_type == "hk2s") {
        stub.hk2s(&cntl, &request, &response, nullptr);
    } else if (lc_convert_type == "hk2t") {
        stub.hk2t(&cntl, &request, &response, nullptr);
    } else if (lc_convert_type == "jp2t") {
        stub.jp2t(&cntl, &request, &response, nullptr);
    } else if (lc_convert_type == "s2hk") {
        stub.s2hk(&cntl, &request, &response, nullptr);
    } else if (lc_convert_type == "s2t") {
        stub.s2t(&cntl, &request, &response, nullptr);
    } else if (lc_convert_type == "s2tw") {
        stub.s2tw(&cntl, &request, &response, nullptr);
    } else if (lc_convert_type == "s2twp") {
        stub.s2twp(&cntl, &request, &response, nullptr);
    } else if (lc_convert_type == "t2hk") {
        stub.t2hk(&cntl, &request, &response, nullptr);
    } else if (lc_convert_type == "t2jp") {
        stub.t2jp(&cntl, &request, &response, nullptr);
    } else if (lc_convert_type == "jp2t") {
        stub.jp2t(&cntl, &request, &response, nullptr);
    } else if (lc_convert_type == "t2s") {
        stub.t2s(&cntl, &request, &response, nullptr);
    } else if (lc_convert_type == "t2tw") {
        stub.t2tw(&cntl, &request, &response, nullptr);
    } else if (lc_convert_type == "tw2s") {
        stub.tw2s(&cntl, &request, &response, nullptr);
    } else if (lc_convert_type == "tw2sp") {
        stub.tw2sp(&cntl, &request, &response, nullptr);
    } else {
      LOG(ERROR) << "Unrecognized convert";
      exit(-1);
    }

    if (!cntl.Failed()) {
        LOG(INFO) << "Received response from " << cntl.remote_side()
            << " to " << cntl.local_side();

        if( response.code() == 0) {
            LOG(INFO) << "success";
            LOG(INFO) << response.message();
        } else {
          LOG(ERROR) << "Fail to receive response";
        }
        LOG(INFO)<< " latency=" << cntl.latency_us() << "us";
    } else {
        LOG(WARNING) << cntl.ErrorText();
    }
    return 0;
}